package belf.migrate.engine.job.cdc;

import belf.migrate.api.common.JobMode;
import belf.migrate.api.job.Job;
import belf.migrate.api.sink.jdbc.JdbcSink;
import belf.migrate.api.source.cdc.CDCSource;
import belf.migrate.api.table.schema.AbstractSinkCatalog;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.taskconf.TaskConf;
import belf.migrate.api.type.LogLevelType;
import belf.migrate.api.util.PostLog;
import belf.migrate.engine.BelfEvn;
import belf.migrate.engine.job.*;
import belf.migrate.engine.job.structure.StructureJobRunner;
import belf.migrate.api.taskconf.Task;
import lombok.extern.slf4j.Slf4j;

import static belf.migrate.api.job.JobCode.*;

/**
 * 默认的异构关系库数据迁移任务：启动{@link CDCSource}捕获源端流式ChangedEvent，启动{@link JdbcSink}插入目标端数据库。<br/>
 * 数据库表结构同步任务参考{@link StructureJobRunner}，不在这个类实现。分开实现以减少代码复杂度。
 */
@Slf4j
public class CDCJobRunner implements JobRunner {
    private final Job job;
    private final Task task;
    private final TaskConf taskConf;

    private JobContext jobContext;

    public CDCJobRunner(Job job) {
        this.job = job;
        this.task = job.getTask();
        this.taskConf = task.getTaskConf();
    }

    public Job getJob() {
        return job;
    }

    public JobContext getJobContext() {
        return jobContext;
    }

    @Override
    public JobResult execute() {
        PostLog.sendLogModel(
                job.getJobId(),
                job.getTask().getTaskType().getName(),
                "同步数据开始！",
                LogLevelType.INFO.getType(),
                EXECUTING,
                jobContext.getProgress());
        JobResult executedResult = new JobResult(EXECUTING, "CDC数据同步Job提交成功。");

        jobContext = new JobContext("" + job.getJobId(), taskConf);
        AbstractSinkCatalog sinkCatalog = CatalogCreator.createSinkCatalog(taskConf);
        if (null == sinkCatalog) {
            log.error("No Sink Catalog for {}, Stop Job {}!",
                    taskConf.getSinkConf().getCatalogName(),
                    job.getJobId());
            executedResult = new JobResult(NO_SOURCE_CATALOG, "提交Job失败，没有找到目标端" +
                    taskConf.getSinkConf().getCatalogName() +
                    "可适配的Catalog类，请联系开发者。");
            return executedResult;
        }
        jobContext.setSinkCatalog(sinkCatalog);
        BelfEvn.getInstance().addJobContext("" + job.getJobId(), jobContext);

        CDCSource cdcSource = CDCSourceCreator.createSource(jobContext, taskConf);
        if (null == cdcSource) {
            log.error("No CDCSource for {}, Stop Job {}!",
                    taskConf.getSourceConf().getCatalogName(),
                    job.getJobId());
            executedResult = new JobResult(NO_SOURCE_CATALOG, "提交Job失败，没有找到源端" +
                    taskConf.getSourceConf().getCatalogName() +
                    "可适配的Catalog类，请联系开发者。");
            return executedResult;
        }
        BelfEvn.getInstance().addCDCSource(cdcSource);

        JdbcSink jdbcSink = JdbcSinkCreator.createSink(jobContext);
        if (null == jdbcSink) {
            log.error("No JdbcSink for {}, Stop Job {}!",
                    taskConf.getSinkConf().getCatalogName(),
                    job.getJobId());
            executedResult = new JobResult(NO_SOURCE_CATALOG, "提交Job失败，没有找到目标端" +
                    taskConf.getSourceConf().getCatalogName() +
                    "可适配的JdbcSink类，请联系开发者。");
            return executedResult;
        }
        BelfEvn.getInstance().addJdbcSink(jdbcSink);

        new CDCSourceThread(job.getJobId(), cdcSource).start();
        new JdbcSinkThread(job.getJobId(), jdbcSink).start();

        //FIXME: 默认返回成功，后续需要完善。
        executedResult.setCode(SUCCESS);
        return executedResult;
    }

    @Override
    public JobMode getMode() {
        return JobMode.STREAMING;
    }

    class CDCSourceThread extends Thread {
        private final CDCSource cdcSource;
        private final long jobId;

        public CDCSourceThread(long jobId, CDCSource cdcSource) {
            this.jobId = jobId;
            this.cdcSource = cdcSource;
            setName("CDCSourceThread-" + jobId);
        }

        @Override
        public void run(){
            cdcSource.start();
        }
    }

    class JdbcSinkThread extends Thread {
        private final JdbcSink jdbcSink;
        private final long jobId;

        public JdbcSinkThread(long jobId, JdbcSink jdbcSink) {
            this.jobId = jobId;
            this.jdbcSink = jdbcSink;
            setName("JdbcSinkThread-" + jobId);
        }

        @Override
        public void run(){
            jdbcSink.start();
        }
    }
}
